#pragma once

#include <rocksdb/comparator.h>
#include <rocksdb/db.h>
#include <rocksdb/options.h>
#include <rocksdb/slice.h>
#include <rocksdb/status.h>
#include <rocksdb/utilities/transaction.h>
#include <rocksdb/utilities/transaction_db.h>

#include <unordered_map>
#include <unordered_set>

#include "redis/store/common.h"
#include "redis/store/scope_lock.h"
#include "redis/store/scope_snapshot.h"
#include "slash/coding.h"

// cf_handles_[0] key score_member_size ZSetsMetaValue
// cf_handles_[1] key_version_member score ZSetsMemberKey
// cf_handles_[2] key_version_score_member "" ZSetsScoreKey
// zsets_score_cf compatator score

class RedisZSets {
 public:
  RedisZSets(const rocksdb::Options &options,
             std::vector<rocksdb::ColumnFamilyHandle *> &cf_handles);
  ~RedisZSets();

  void SetTranscationDB(
      const std::shared_ptr<rocksdb::TransactionDB> &transcation_db) {
    transcation_db_ = transcation_db;
  }

  rocksdb::Status Open(
      rocksdb::TransactionDB *txn_db, rocksdb::Status s,
      std::vector<rocksdb::ColumnFamilyDescriptor> &column_families);
  rocksdb::Status ZAdd(
      const rocksdb::Slice &key, const std::vector<ScoreMember> &score_members,
      int32_t *ret,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZCard(
      const rocksdb::Slice &key, int32_t *card,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZCount(
      const rocksdb::Slice &key, double min, double max, bool left_close,
      bool right_close, int32_t *ret,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZRange(
      const rocksdb::Slice &key, int32_t start, int32_t stop,
      std::vector<ScoreMember> *score_members,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZIncrby(
      const rocksdb::Slice &key, const rocksdb::Slice &member, double increment,
      double *ret,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZRank(
      const rocksdb::Slice &key, const rocksdb::Slice &member, int32_t *rank,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
  rocksdb::Status ZScan(const Slice &key, int64_t cursor,
                        const std::string &pattern, int64_t count,
                        std::vector<ScoreMember> *score_members,
      int64_t *next_cursor,
      const std::shared_ptr<rocksdb::Transaction> &transcation = nullptr);
 private:
  RedisZSets(const RedisZSets &);
  void operator=(const RedisZSets &);

  std::shared_ptr<rocksdb::TransactionDB> transcation_db_;
  rocksdb::Options options_;
  std::vector<rocksdb::ColumnFamilyHandle *> &cf_handles_;
  rocksdb::WriteOptions default_write_options_;
  rocksdb::ReadOptions default_read_options_;
  rocksdb::CompactRangeOptions default_compact_range_options_;
  rocksdb::TransactionOptions default_transaction_options_;
  std::shared_ptr<ScopeLockMgr> scope_lock_mgr_;
};
